package com.zshstart.app;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import com.zshstart.func.MyStringDebeziumDeserializationSchema;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 自定义反序列器
 * @author zshstart
 * @create 2021-09-22 19:45
 */
public class FlinkCDCWithDataStreamWithDeserialization {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /**
        //1.1 指定状态后端并开启ck
        //1.1.1指定存储checkPoint的位置
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flink-cdc-2021/ck"));
        //1.1.2开启checkpoint
        env.enableCheckpointing(5000L);
        //1.1.3设置checkpoint超时时间
        env.getCheckpointConfig().setCheckpointTimeout(1000L);
        //1.1.4设置最大同时存在的checkpoint
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //1.1.5设置最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        */

        //2.创建mysql cdc source 并创建流
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("gmall-flink-2021")
                .tableList("gmall-flink-2021.base_trademark")
                .startupOptions(StartupOptions.initial())
                .deserializer(new MyStringDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> dataStreamSource = env.addSource(sourceFunction);
        //3.打印数据
        dataStreamSource.print();
        //4.启动任务
        env.execute("FlinkCDCWithDataStream");
    }
}
